package com.project.KafkaRoute.executor;

import com.project.MessageRoute.executor.MessageRouteSetting;
import com.project.MessageRoute.executor.ProductExecutor;
import com.project.MessageRoute.model.MessageRouteBean;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.util.List;

@Component
public class KafkaListenerConsumer implements KafkaListenerConcurrently{

    @Autowired
    private ProductExecutor productExecutor;

    @Override
    public Boolean consumeMessage(ConsumerRecords<String, String> poll) {

        for (ConsumerRecord<String, String> msg : poll) {
            try {
                String in_str = msg.value();
                String name = msg.topic();
                List<MessageRouteBean> bean = MessageRouteSetting.getBean(name);
                for (MessageRouteBean item : bean){
                    System.out.println("===============>listener message: " + in_str);
                    if (item.isAsync()){
                        item.syncInvoke(in_str).subscribe((json) -> productExecutor.sendGroup(item.getRouteMethod().groupId(), item.getRouteMethod().topic(), item.getResult(), json));
                    } else {
                        productExecutor.sendGroup(item.getRouteMethod().groupId(), item.getRouteMethod().topic(), item.getResult(), item.invoke(in_str));
                    }
                    System.out.println("===============>end listener message");
                }
            } catch (Exception e) {
                e.printStackTrace();
                return false;
            }
        }

        return true;
    }
}
